package com.example.mq.server.dao;

import com.example.mq.common.MqException;
import com.example.mq.server.core.Binding;
import com.example.mq.server.core.Exchange;
import com.example.mq.server.core.MSGQueue;
import com.example.mq.server.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

public class MemoryDataCenter {
    // 交换机管理map，key是交换机名字 value是交换机
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // 队列管理map，key是队列名字，value是队列
    private final ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 交换机绑定队列 第一个key为交换机名字 第二个key是队列名称
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 消息管理map
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // 队列中的消息
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 表示未被确认的的消息
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAck = new ConcurrentHashMap<>();

    // 交换机的相关操作
    public void addExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 交换机添加成功！exchangeName = " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName = " + exchangeName);
    }

    // 队列的相关操作
    public void addQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 队列添加成功！queueName = " + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName = " + queueName);
    }

    // 绑定的相关操作
    public void addBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机从未绑定则创建一个绑定关系
            bindingMap = new ConcurrentHashMap<>();
            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }
        // 此方法是没有则创建一个
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
//                k -> new ConcurrentHashMap<>());
        // 再查一下是否有队列绑定信息
        // 保证操作原子性，加锁
        synchronized(bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                // 有绑定信息则不能绑定
                throw new MqException("[MemoryDataCenter] 绑定已经存在！exchangeName = " + binding.getExchangeName()
                        + ",queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 交换机绑定队列成功！exchangeName = " + binding.getExchangeName() +
                    ",queueName = " + binding.getQueueName());
        }
    }

    // 获取一个绑定关系
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    // 获取所有绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 删除绑定关系
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有任何绑定，不能删除
            throw new MqException("[MemoryDataCenter] exchangeName = " + binding.getExchangeName() + "交换机无绑定，无法删除！");
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 交换机解绑队列成功！exchangeName = " + binding.getExchangeName() +
                ",queueName = " + binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功！ messageId = " + message.getMessageId());
    }

    // 根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 删除消息
    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功！ messageId = " + messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if (messages == null) {
//            // 如果队列中无消息先创建一个链表用于接收消息
//            messages = new LinkedList<>();
//            queueMessageMap.put(queue.getName(), messages);
//        }
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        // 添加消息
        synchronized(messages) {
            messages.add(message);
            addMessage(message);
        }
        System.out.println("[MemoryDataCenter] 消息发送到队列成功！ queueName = " + queue.getName() +
                ",messageId = " + message.getMessageId());
    }

    // 队列中取消息给消费者
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized(messages) {
            if (messages.size() == 0) {
                return null;
            }
            Message message = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息消费成功！queueName = " + queueName +
                    ",messageId = " + message.getMessageId());
            return message;
        }
    }

    // 获取队列中消息个数
    public int getQueueMessageNum(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized(messages) {
            return messages.size();
        }
    }

    // 添加未确认的消息，等消费者反馈再删除消息的一种机制
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列！ messageId = " + message.getMessageId());
    }

    // 删除未确认的消息（已确认）
    public void deleteMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.get(queueName);
        if (messageMap == null || messageMap.size() == 0) {
            return;
        }
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除！ messageId = " + messageId);
    }

    // 获取指定的未确认消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAck.get(queueName);
        if (messageMap == null || messageMap.size() == 0) {
            return null;
        }
        return messageMap.get(messageId);
    }

    // 当内存数据丢失，从硬盘中恢复数据
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 清空内存中所有的数据，防止数据残留
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1.恢复交换机的数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 2.恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 3.恢复绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 4.恢复所有的消息数据,每个队列有每个队列的消息
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessagesFromQueue(queue.getName());
            // 将每个队列对应的消息恢复
            queueMessageMap.put(queue.getName(),messages);
            // 恢复消息管理中的所有消息
            for(Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
    }
}
